#include <netdb.h> 
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <stdio.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <stdlib.h>
#include <iostream>
#include <string>
#include <vector>
#include <pthread.h>
#include <sstream>
#include <fstream>
#include <signal.h>
#include "sequencer.h"
#define NUM_THREADS 1

 

using namespace std;
pthread_mutex_t mutexwrite;
pthread_mutex_t mutexIPadd;

//listens to messages and sends them in order to each client
vector<Message>Sequencer :: seqMessageQueue;
clientMessageSequencerMap Sequencer ::clientMsgSeqMap;

int Sequencer :: currentSequenceNumber = 0;
int Sequencer :: sockSequencer;

void * Sequencer :: callForListening(void * data){
  listeningThData * dataReceived = (listeningThData *) data;
  listen_and_sequence(dataReceived->portNum, dataReceived->callType);
}

void Sequencer::listen_and_sequence(int portNum, int callType){
  
  int bytes_read; // <- note how this is now on its own line!
  socklen_t addr_len; // <- and this too, with a different type.
  char recv_data[MAX_MSGSIZE];
  
  struct sockaddr_in server_addr , client_addr;
  
  char * ipbuf;
  char hostbuf[256];
  
  struct hostent *hostentry;
  int ret;
  
  ret = gethostname(hostbuf,sizeof(hostbuf));
  
  if(ret == -1){
    perror("gethostname");
    exit(1);
  }
  
  hostentry = gethostbyname(hostbuf);
  
  if(hostentry == NULL){
    perror("gethostbyname");
    exit(1);
  }
  ipbuf = inet_ntoa(*((struct in_addr *)hostentry->h_addr_list[0]));
  
  if(ipbuf == NULL){
    perror("inet_ntoa");
    exit(1);
  }
  
  cout<<"IP address:"<<ipbuf<<endl;
  
  if ((sockSequencer = socket(AF_INET, SOCK_DGRAM, 0)) == -1) {
    perror("Socket");
    exit(1);
  }
  int a=MAX_MSGSIZE;
  ret=setsockopt(sockSequencer, SOL_SOCKET, SO_RCVBUF, &a, sizeof(int));
  if(ret == -1){
    perror("sock recv buf error");
    exit(1);
  }
  
  server_addr.sin_family = AF_INET;
  server_addr.sin_port = htons(portNum);
  server_addr.sin_addr.s_addr = INADDR_ANY;
  bzero(&(server_addr.sin_zero),8);
  
  
  if (bind(sockSequencer,(struct sockaddr *)&server_addr,
	   sizeof(struct sockaddr)) == -1)
    {
      perror("Bind");
      exit(1);
    }
  
  addr_len = sizeof(struct sockaddr);
  
  cout << "\nUDPServer Waiting for client " << portNum;
  fflush(stdout);

  if(callType == 1)
    {
      Message reboot;
      reboot.messagetype = MSG_REBOOT;
      reboot.message = ipbuf;
      broadcast(reboot);
    }
  
  
  while (1)
    { 
      bytes_read = recvfrom(sockSequencer,recv_data,MAX_MSGSIZE,0,
			    (struct sockaddr *)&client_addr, &addr_len);
      
      recv_data[bytes_read] = '\0';
      pthread_mutex_lock(&mutexIPadd);
      string ipAdd = inet_ntoa(client_addr.sin_addr);
      //cout << "Message from IP : " << ipAdd <<endl;
       
      string temp;	
      istringstream ss (recv_data);
      Message m;
      int i=0, seq=0,type=0,instanceNum =0;
      string strmsg,struname;
      
      while(getline(ss,temp,'|'))
	{
      	  switch(i){
      	  case 0:
      	  seq=atoi(temp.c_str());
      	  break;
      	  case 1:
      	  struname=temp;
      	  break;
      	  case 2:
      	  strmsg=temp;
      	  break;
      	  case 3:
      	  instanceNum = atoi(temp.c_str());
      	  break;
      	  case 4:
      	  type=atoi(temp.c_str());
      	  break;
      	  }
      	  i++;
      	  
      }
      m.seqno=seq;
      m.username=struname;
      m.message=strmsg;
      m.messagetype=type;
      m.instancenumber = instanceNum;
      
      
      int flag=0;
      string ip=inet_ntoa(client_addr.sin_addr);
      // cout << "IP TABLE size: " <<iptables.size() << endl;  
      
      if(iptables.size()>0){
	
	for(int i=0;i<iptables.size();i++)
	  {
	    if(iptables[i].compare(ip)==0)
	      {
		flag=1;
		break;
	      }
	  }
	if(flag==0)
	  {
	   
	    clientMsgSeqMap[ip] = 0;
	    iptables.push_back(inet_ntoa(client_addr.sin_addr));
	    for(int ind =0; ind < iptables.size(); ind++)
	      {
		Message MSGaskClientsToAddIP;
		MSGaskClientsToAddIP.messagetype = MSG_ADDIPTOTABLE;
		MSGaskClientsToAddIP.message = iptables[ind];
		broadcast(MSGaskClientsToAddIP);
	      }
	  }
      }
      else
	{
	 
	  clientMsgSeqMap[ip] = 0;
	  iptables.push_back(inet_ntoa(client_addr.sin_addr));
	  for(int ind =0; ind < iptables.size(); ind++)
	    { 
	      Message MSGaskClientsToAddIP;
	      MSGaskClientsToAddIP.messagetype = MSG_ADDIPTOTABLE;
	      MSGaskClientsToAddIP.message = iptables[ind];
	      broadcast(MSGaskClientsToAddIP);
	    }
	}
      //      cout << iptables.size();
      // for(int i = 0; i< iptables.size(); i++)
      //cout << i <<" : "<< iptables[i] <<endl;
      
      if(m.messagetype == MSG_CLIENTCHECKSALIVE)
	{
	  //Reply to this specific client with its instancenumber and messagetype MSG_SERVERCHECKSALIVE
	  // cout <<"Client - I am alive : "<<m.instancenumber <<endl;
	  Message replyClient;
	  replyClient.messagetype = MSG_SERVERCHECKSALIVE;
	  char* argv[3];
	  argv[0]="abcd\0";
	  string str = ip;
	  //  cout <<"This message was intended for : "<< m.message<<endl;
	  //cout <<"Server says, I am sending to the IP : "<<ip <<endl;
	  char * writable = new char[str.size() + 1];
	  std::copy(str.begin(), str.end(), writable);
	  writable[str.size()] = '\0';
	  argv[1]=writable;
	  char num[5];
	  sprintf(num, "%d", chatport);
	  argv[2]=num;
	  
	  replyClient.instancenumber = m.instancenumber;
	  stringstream temp(stringstream::in | stringstream::out);
	  temp<<replyClient;
	  send(argv,temp.str().c_str());
	  delete[] writable;
	  
	  //Send back message to the client with this IP that it is alive!
	  // cout <<"Client "<<ip <<" checks if sequencer is alive!!!!!" <<endl;
	}
	else if(m.messagetype == MSG_REMOVEFROMIPTABLE)
	{
		//cout <<"PRINT IP TABLE : "<< iptables.size() <<endl;
      		//for(int opop = 0; opop< iptables.size(); opop++)
      		//	cout << opop <<" : "<< iptables[opop] <<endl;

           cout << "Removing IP : " << m.message <<endl;
	   for(int j =0; j < iptables.size(); j++)
	   {
	      if(iptables[j].compare(m.message) == 0)
		{
		cout << "Successfully removed IP : " << m.message <<endl;
		  
		  iptables.erase(iptables.begin() + j);
		  break;
		}
	    }
		//cout <<"COMES HEREEEEEEEEEEEE!!! "<<endl;
	
	    Message MSGaskClientsToRemIP;
	    MSGaskClientsToRemIP.messagetype = MSG_REMOVEFROMIPTABLE;
	    MSGaskClientsToRemIP.message = m.message;
	    broadcast(MSGaskClientsToRemIP);

	//cout <<"End of remove from iptable" <<endl;
	}
      else if(m.messagetype == MSG_DATAMSG)
	{
	  //DATA MESSAGE
	 // cout<<"Message number seen by sequencer = "<<currentSequenceNumber<<"."<<"Message sequence number : "<<m.seqno <<endl;
	  //Sequencer receives messages from client. Assigns a sequence number and sends it to every one in the chat
	  if(m.seqno == clientMsgSeqMap[ip])
	    {

//	      cout << "currnet Msg seq for Map : "<< clientMsgSeqMap[ip] <<endl;
	      
	      clientMsgSeqMap[ip] = clientMsgSeqMap[ip] + 1;
	      m.seqno = currentSequenceNumber++;
	      Sequencer::seqMessageQueue.push_back(m);

		//LOGGING  this to  a file
		

	      std::fstream fb("logFile.txt", std::ios::in | std::ios::out | std::ios::app );
	      string saysTemp = " says : ";
	      string endTemp = "\n";
	      fb << m.username << saysTemp << m.message << endTemp;

             // fwrite(recv_data, sizeof(recv_data[0]), bytes_read + 1 , pFile);
             // fclose(pFile);
	
	      
	      cout<<m.username<<" says : ";
	      cout<<m.message<<endl;
	      
	      fflush(stdout);
	      broadcast(m);
	    }
	  else
	    {
	      //Ask client to retransmit all these messages including the current one again
	      for(int idTmp = clientMsgSeqMap[ip]; idTmp <=  m.seqno; idTmp++)
		{
		  Message reTrnsmit;
		  reTrnsmit.messagetype = MSG_ASKCLIENTRETRANSMIT;
		  reTrnsmit.seqno = idTmp;
		  char* argv[3];
		  argv[0]="abcd\0";
		  string str = ip;
		  char * writable = new char[str.size() + 1];
		  std::copy(str.begin(), str.end(), writable);
		  writable[str.size()] = '\0';
		  argv[1]=writable;
		  char num[5];
		  sprintf(num, "%d", chatport);
		  argv[2]=num;
		  stringstream temp(stringstream::in | stringstream::out);
		  temp<<reTrnsmit;
		  send(argv,temp.str().c_str());
		  delete[] writable;  
		}
	    }
	}
      else if(m.messagetype == MSG_RETRANSMITDATA)
	{
	  //REQUEST FOR MESSAGE TO BE RETRANSMITTED
	  int indexVal = 0;
	  for (indexVal = 0; indexVal < seqMessageQueue.size(); indexVal++)
	    {
	      if(seqMessageQueue[indexVal].seqno == m.seqno)
		break;
	    }
	  
	  if(indexVal == seqMessageQueue.size())
	    {
	    //  cout << "Alert!!Client might be have transmitted incorrect sequence number." <<endl;
	    }
	  
	  char* argv[3];
          argv[0]="abcd\0";
          string str = ip;
          char * writable = new char[str.size() + 1];
	  std::copy(str.begin(), str.end(), writable);
          writable[str.size()] = '\0';
          argv[1]=writable;
          char num[5];
          sprintf(num, "%d", chatport);
          argv[2]=num;
	 // cout << "RE - Transmitting message with sequence number :  " << m.seqno << endl;
	//  cout << "Message was : " <<seqMessageQueue[indexVal].message <<endl;
	  
          
          stringstream temp(stringstream::in | stringstream::out);
          temp<<seqMessageQueue[indexVal];
          send(argv,temp.str().c_str());
          delete[] writable;
	  
	}
      else
	{
	 // cout << "Message Type undefined : " << m.messagetype<<endl;
	}
      
      pthread_mutex_unlock(&mutexIPadd);
    }
  
}

/*void Sequencer:: handlingMethod(int opopop)
{
alarm(1);
 cout <<"call for checking if clients are alive" <<endl;
}

void * Sequencer:: callForChecking(void * data)
{
 
  signal( SIGALRM, &Sequencer::handlingMethod);
  alarm(1);
	
	
}
*/
void * Sequencer :: callForListeningUI(void * data)
{
 // cout <<"LISTENING THREAD INITIATED!!!!!!! "<<endl;
  // cout <<"Sequence number : " <<currentSequenceNumber <<endl;
	
  while(1)  
    {
     
     // cout <<"whith in teh FOR LOOOOOOOOOOOOOOOOOOOOOOOPPPPPPPPPPPPPPPPPP1!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!" <<endl; 
      char* argv[2];
      argv[0]="abcd\0";
      char num[5];
      sprintf(num, "%d", chatport);
      argv[1]=num;
      char * ipbuf=getIPAddress();
      char send_data[MSG_SIZE];
     // cout<<"Initiating mutexwrite lock! "<<endl;
      //      pthread_mutex_lock(&mutexwrite);
      cout<<"Type Something (q or Q to quit):"<<endl;
      cin.getline(send_data, MSG_SIZE);
      if(feof(stdin))
	{
	  // pthread_mutex_unlock(&mutexwrite);
	    break;  
	}
      if ((strcmp(send_data , "q") == 0) || strcmp(send_data , "Q") == 0)
	{
	  pthread_mutex_unlock(&mutexwrite);
	  break;
	}
      Message m;
    //  cout <<"Current Sequencer number : " <<currentSequenceNumber;
      

      // m.seqno = DistChat :: currentClientSequenceNumber++;
      m.messagetype = MSG_DATAMSG;
      m.username=userName;
      m.message=send_data;
  //    cout <<"Sequence number : " <<currentSequenceNumber <<endl;
      m.seqno = currentSequenceNumber++;
	//char * ipbuf=getIPAddress();

//	cout <<"printing done by : " <<ipbuf<<endl;
      std::fstream fb("logFile.txt", std::ios::in | std::ios::out | std::ios::app );
      string saysTemp = " says : ";
      string endTemp = "\n";
      fb << m.username << saysTemp << m.message << endTemp;

      // DistChat::sentMessages.push_back(m);

      /*TESTING CODE. DON'T FORGET TO comment THIS !!!!!!!!!!!!!! */
      /*      if(currentSequenceNumber == 5)
	      currentSequenceNumber = currentSequenceNumber + 2;*/
	

      cout<<userName<<" says : "<<send_data<<endl;
      
      
      seqMessageQueue.push_back(m);
      //          pthread_mutex_unlock(&mutexwrite);
      broadcast(m);
    }
  //rewind(stdin);
 // cout <<"end of listening thread!!! "<<endl;
  
}

Sequencer :: Sequencer(char * argv[])
{
  /* Message tester;
     tester.seqno = 5;
     tester.message = "Fuck you guys!!";
     tester.username = "Anand";
     seqMessageQueue.push_back(tester);
     
     Message tester2;
     tester2.seqno = 6;
     tester2.message = "Screw you guys!";
     tester2.username = "Anand";
     seqMessageQueue.push_back(tester2);*/
  
  DistChat ::currentClientSequenceNumber = 0;
  DistChat::sentMessages.clear();
  currentSequenceNumber = 0;
  pthread_t threadSequencerListens;
  pthread_t threadSequencerListensUI;
  pthread_t threadSequencerChecksClient;

  listeningThData dataToSend;
  strcpy( userName,argv[1]);
  dataToSend.portNum= atoi(argv[2]);
  dataToSend.callType = 0;
  chatport=atoi(argv[2]);
  
  int tSl = pthread_create(&threadSequencerListens, NULL, &Sequencer::callForListening, (void *)&dataToSend);
  int tUi = pthread_create(&threadSequencerListensUI, NULL, &Sequencer::callForListeningUI, (void *)&dataToSend);

	pthread_join(threadSequencerListensUI, NULL);
  
  if(tSl){
    printf("ERROR; return code from pthread_create() is %d\n", tSl);        
    exit(-1);
  }
}
   //new constructor
Sequencer::Sequencer(char *argv[], vector<string> iptables){
  //assign iptables
  DistChat::currentClientSequenceNumber = 0;
  DistChat::sentMessages.clear();
  DistChat::currIPtable = iptables;
  pthread_t threadSequencerListens;
  pthread_t threadSequencerListensUI;
  
  listeningThData dataToSend;
  strcpy( userName,argv[1]);
	//here port arg = 3 because we will reuse argv from client invocation in dchat main()
  dataToSend.portNum= atoi(argv[3]);
  dataToSend.callType = 1;
  chatport=atoi(argv[3]);
  cout <<"Listening on port : " << chatport <<endl;
  int tSl = pthread_create(&threadSequencerListens, NULL, &Sequencer::callForListening, (void *)&dataToSend);
  int tUi = pthread_create(&threadSequencerListensUI, NULL, &Sequencer::callForListeningUI, (void *)&dataToSend);
  pthread_join(threadSequencerListensUI, NULL);
  
  if(tSl){
    printf("ERROR; return code from pthread_create() is %d\n", tSl);        
    exit(-1);
  }

   
}

Sequencer :: ~Sequencer(){
  close(sockSequencer);
}
